package com.chukun.multi.twophase;

import com.chukun.multi.enums.AlarmType;
import com.chukun.multi.guarded.example.AlarmAgent;
import com.chukun.multi.guarded.example.AlarmInfo;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author chukun
 *  报警线程
 */
public class AlarmSendingThread extends AbstractTerminatableThread {


    private AlarmAgent alarmAgent = new AlarmAgent();

    // 告警队列
    private final BlockingQueue<AlarmInfo> alarmQueue;
    private final ConcurrentMap<String, AtomicInteger> submittedAlarmRegistry;

    public AlarmSendingThread() {
        alarmQueue = new ArrayBlockingQueue<>(1024);
        submittedAlarmRegistry = new ConcurrentHashMap<>();
        alarmAgent.init();
    }


    @Override
    protected void doRun() throws Exception {
        AlarmInfo alarmInfo = alarmQueue.take();
        terminationToken.reservations.decrementAndGet();
        try {
            // 将报警信息，发送到报警服务器
            alarmAgent.sendAlarm(alarmInfo);
        } catch (Exception e) {
            e.printStackTrace();
        }

        /**
         * 处理恢复报警: 将相应的报警从注册表中删除，使得相应故障恢复后若再次出现相同的报警，
         * 可以发送到报警服务器
         */
        if (AlarmType.RESUME == alarmInfo.type) {
            String key = AlarmType.FAULT.toString() + ":" + alarmInfo.getId() + "@" + alarmInfo.getExtraInfo();
            submittedAlarmRegistry.remove(key);
            key = AlarmType.RESUME.toString() + ":" + alarmInfo.getId() + "@" + alarmInfo.getExtraInfo();
            submittedAlarmRegistry.remove(key);
        }
    }

    public int sendAlarm(final AlarmInfo alarmInfo) {
        AlarmType type = alarmInfo.type;
        String id = alarmInfo.getId();
        String extraInfo = alarmInfo.getExtraInfo();
        if (terminationToken.isToShutdown()) {
            // 记录报警
            System.err.println("rejected alarm : " + id + "," + extraInfo);
            return -1;
        }
        int duplicateSubmissionCount = 0;
        try {
            AtomicInteger preSubmittedCounter;
            preSubmittedCounter = submittedAlarmRegistry.putIfAbsent(type.toString() + ":" + id + "@" + extraInfo, new AtomicInteger(0));
            if (null == preSubmittedCounter) {
                terminationToken.reservations.incrementAndGet();
                alarmQueue.put(alarmInfo);
            } else {
                // 故障为恢复，不用重复发送报警信息给服务器，仅增加计数
                duplicateSubmissionCount = preSubmittedCounter.incrementAndGet();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return duplicateSubmissionCount;
    }

    @Override
    protected void doCleanup(Exception ex) {
        if (null != ex && !(ex instanceof InterruptedException)) {
            ex.printStackTrace();
        }
        alarmAgent.disconnect();
    }
}
